/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUMixStreamRecord<?>WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bff.gaia.mix.runtime.streamrecord;

import com.bff.gaia.annotation.Internal;
import com.bff.gaia.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
import com.bff.gaia.api.common.typeutils.CompositeTypeSerializerSnapshot;
import com.bff.gaia.api.common.typeutils.CompositeTypeSerializerUtil;
import com.bff.gaia.api.common.typeutils.TypeSerializer;
import com.bff.gaia.api.common.typeutils.TypeSerializerSchemaCompatibility;
import com.bff.gaia.core.memory.DataInputView;
import com.bff.gaia.core.memory.DataOutputView;
import com.bff.gaia.runtime.jobgraph.OperatorID;
import com.bff.gaia.mix.api.watermark.Watermark;
import com.bff.gaia.mix.runtime.streamstatus.MixStreamStatus;

import java.io.IOException;

import static java.util.Objects.requireNonNull;

/**
 * Serializer for {@link MixStreamRecord}, {@link Watermark}, {@link LatencyMarker}, and
 * {@link MixStreamStatus}.
 *
 * <p>This does not behave like a normal {@link TypeSerializer}, instead, this is only used at the
 * stream task/operator level for transmitting MixStreamRecords and Watermarks.
 *
 * @param <T> The type of value in the MixStreamRecord
 */
@Internal
public final class MixStreamElementSerializer<T> extends TypeSerializer<MixStreamElement> {

	private static final long serialVersionUID = 1L;

	private static final int TAG_REC_WITH_TIMESTAMP = 0;
	private static final int TAG_REC_WITHOUT_TIMESTAMP = 1;
	private static final int TAG_WATERMARK = 2;
	private static final int TAG_LATENCY_MARKER = 3;
	private static final int TAG_STREAM_STATUS = 4;


	private final TypeSerializer<T> typeSerializer;

	public MixStreamElementSerializer(TypeSerializer<T> serializer) {
		if (serializer instanceof MixStreamElementSerializer) {
			throw new RuntimeException("MixStreamRecordSerializer given to MixStreamRecordSerializer as value TypeSerializer: " + serializer);
		}
		this.typeSerializer = requireNonNull(serializer);
	}

	public TypeSerializer<T> getContainedTypeSerializer() {
		return this.typeSerializer;
	}

	// ------------------------------------------------------------------------
	//  Utilities
	// ------------------------------------------------------------------------

	@Override
	public boolean isImmutableType() {
		return false;
	}

	@Override
	public MixStreamElementSerializer<T> duplicate() {
		TypeSerializer<T> copy = typeSerializer.duplicate();
		return (copy == typeSerializer) ? this : new MixStreamElementSerializer<T>(copy);
	}

	// ------------------------------------------------------------------------
	//  Utilities
	// ------------------------------------------------------------------------

	@Override
	public MixStreamRecord<T> createInstance() {
		return new MixStreamRecord<T>(typeSerializer.createInstance());
	}

	@Override
	public int getLength() {
		return -1;
	}

	@Override
	public MixStreamElement copy(MixStreamElement from) {
		// we can reuse the timestamp since Instant is immutable
		if (from.isRecord()) {
			MixStreamRecord<T> fromRecord = from.asRecord();
			return fromRecord.copy(typeSerializer.copy(fromRecord.getValue()));
		}
		else if (from.isWatermark() || from.isMixStreamStatus() || from.isLatencyMarker()) {
			// is immutable
			return from;
		}
		else {
			throw new RuntimeException();
		}
	}

	@Override
	public MixStreamElement copy(MixStreamElement from, MixStreamElement reuse) {
		if (from.isRecord() && reuse.isRecord()) {
			MixStreamRecord<T> fromRecord = from.asRecord();
			MixStreamRecord<T> reuseRecord = reuse.asRecord();

			T valueCopy = typeSerializer.copy(fromRecord.getValue(), reuseRecord.getValue());
			fromRecord.copyTo(valueCopy, reuseRecord);
			return reuse;
		}
		else if (from.isWatermark() || from.isMixStreamStatus() || from.isLatencyMarker()) {
			// is immutable
			return from;
		}
		else {
			throw new RuntimeException("Cannot copy " + from + " -> " + reuse);
		}
	}

	@Override
	public void copy(DataInputView source, DataOutputView target) throws IOException {
		int tag = source.readByte();
		target.write(tag);

		if (tag == TAG_REC_WITH_TIMESTAMP) {
			// move timestamp
			target.writeLong(source.readLong());
			typeSerializer.copy(source, target);
		}
		else if (tag == TAG_REC_WITHOUT_TIMESTAMP) {
			typeSerializer.copy(source, target);
		}
		else if (tag == TAG_WATERMARK) {
			target.writeLong(source.readLong());
		}
		else if (tag == TAG_STREAM_STATUS) {
			target.writeInt(source.readInt());
		}
		else if (tag == TAG_LATENCY_MARKER) {
			target.writeLong(source.readLong());
			target.writeLong(source.readLong());
			target.writeLong(source.readLong());
			target.writeInt(source.readInt());
		} else {
			throw new IOException("Corrupt stream, found tag: " + tag);
		}
	}

	@Override
	public void serialize(MixStreamElement value, DataOutputView target) throws IOException {
		if (value.isRecord()) {
			MixStreamRecord<T> record = value.asRecord();

			if (record.hasTimestamp()) {
				target.write(TAG_REC_WITH_TIMESTAMP);
				target.writeLong(record.getTimestamp());
			} else {
				target.write(TAG_REC_WITHOUT_TIMESTAMP);
			}
			typeSerializer.serialize(record.getValue(), target);
		}
		else if (value.isWatermark()) {
			target.write(TAG_WATERMARK);
			target.writeLong(value.asWatermark().getTimestamp());
		}
		else if (value.isMixStreamStatus()) {
			target.write(TAG_STREAM_STATUS);
			target.writeInt(value.asMixStreamStatus().getStatus());
		}
		else if (value.isLatencyMarker()) {
			target.write(TAG_LATENCY_MARKER);
			target.writeLong(value.asLatencyMarker().getMarkedTime());
			target.writeLong(value.asLatencyMarker().getOperatorId().getLowerPart());
			target.writeLong(value.asLatencyMarker().getOperatorId().getUpperPart());
			target.writeInt(value.asLatencyMarker().getSubtaskIndex());
		}
		else {
			throw new RuntimeException();
		}
	}

	@Override
	public MixStreamElement deserialize(DataInputView source) throws IOException {
		int tag = source.readByte();
		if (tag == TAG_REC_WITH_TIMESTAMP) {
			long timestamp = source.readLong();
			return new MixStreamRecord<T>(typeSerializer.deserialize(source), timestamp);
		}
		else if (tag == TAG_REC_WITHOUT_TIMESTAMP) {
			return new MixStreamRecord<T>(typeSerializer.deserialize(source));
		}
		else if (tag == TAG_WATERMARK) {
			return new Watermark(source.readLong());
		}
		else if (tag == TAG_STREAM_STATUS) {
			return new MixStreamStatus(source.readInt());
		}
		else if (tag == TAG_LATENCY_MARKER) {
			return new LatencyMarker(source.readLong(), new OperatorID(source.readLong(), source.readLong()), source.readInt());
		}
		else {
			throw new IOException("Corrupt stream, found tag: " + tag);
		}
	}

	@Override
	public MixStreamElement deserialize(MixStreamElement reuse, DataInputView source) throws IOException {
		int tag = source.readByte();
		if (tag == TAG_REC_WITH_TIMESTAMP) {
			long timestamp = source.readLong();
			T value = typeSerializer.deserialize(source);
			MixStreamRecord<T> reuseRecord = reuse.asRecord();
			reuseRecord.replace(value, timestamp);
			return reuseRecord;
		}
		else if (tag == TAG_REC_WITHOUT_TIMESTAMP) {
			T value = typeSerializer.deserialize(source);
			MixStreamRecord<T> reuseRecord = reuse.asRecord();
			reuseRecord.replace(value);
			return reuseRecord;
		}
		else if (tag == TAG_WATERMARK) {
			return new Watermark(source.readLong());
		}
		else if (tag == TAG_LATENCY_MARKER) {
			return new LatencyMarker(source.readLong(), new OperatorID(source.readLong(), source.readLong()), source.readInt());
		}
		else {
			throw new IOException("Corrupt stream, found tag: " + tag);
		}
	}

	// ------------------------------------------------------------------------
	//  Utilities
	// ------------------------------------------------------------------------

	@Override
	public boolean equals(Object obj) {
		if (obj instanceof MixStreamElementSerializer) {
			MixStreamElementSerializer<?> other = (MixStreamElementSerializer<?>) obj;

			return typeSerializer.equals(other.typeSerializer);
		} else {
			return false;
		}
	}

	@Override
	public int hashCode() {
		return typeSerializer.hashCode();
	}

	// --------------------------------------------------------------------------------------------
	// Serializer configuration snapshotting & compatibility
	//
	// This serializer may be used by Gaia internal operators that need to checkpoint
	// buffered records. Therefore, it may be part of managed state and need to implement
	// the configuration snapshot and compatibility methods.
	// --------------------------------------------------------------------------------------------

	@Override
	public MixStreamElementSerializerSnapshot<T> snapshotConfiguration() {
		return new MixStreamElementSerializerSnapshot<>(this);
	}

	/**
	 * Configuration snapshot specific to the {@link MixStreamElementSerializer}.
	 * @deprecated see {@link MixStreamElementSerializerSnapshot}.
	 */
	@Deprecated
	public static final class MixStreamElementSerializerConfigSnapshot<T> extends CompositeTypeSerializerConfigSnapshot<MixStreamElement> {

		private static final int VERSION = 1;

		/** This empty nullary constructor is required for deserializing the configuration. */
		public MixStreamElementSerializerConfigSnapshot() {}

		@Override
		public int getVersion() {
			return VERSION;
		}

		@Override
		public TypeSerializerSchemaCompatibility<MixStreamElement> resolveSchemaCompatibility(TypeSerializer<MixStreamElement> newSerializer) {
			return CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot(
				newSerializer,
				new MixStreamElementSerializerSnapshot<>(),
				getSingleNestedSerializerAndConfig().f1);
		}
	}

	/**
	 * Configuration snapshot specific to the {@link MixStreamElementSerializer}.
	 */
	public static final class MixStreamElementSerializerSnapshot<T>
		extends CompositeTypeSerializerSnapshot<MixStreamElement, MixStreamElementSerializer<T>> {

		private static final int VERSION = 2;

		@SuppressWarnings("WeakerAccess")
		public MixStreamElementSerializerSnapshot() {
			super(MixStreamElementSerializer.class);
		}

		MixStreamElementSerializerSnapshot(MixStreamElementSerializer<T> serializerInstance) {
			super(serializerInstance);
		}

		@Override
		protected int getCurrentOuterSnapshotVersion() {
			return VERSION;
		}

		@Override
		protected TypeSerializer<?>[] getNestedSerializers(MixStreamElementSerializer<T> outerSerializer) {
			return new TypeSerializer[]{outerSerializer.getContainedTypeSerializer()};
		}

		@Override
		protected MixStreamElementSerializer<T> createOuterSerializerWithNestedSerializers(TypeSerializer<?>[] nestedSerializers) {
			@SuppressWarnings("unchecked")
			TypeSerializer<T> casted = (TypeSerializer<T>) nestedSerializers[0];

			return new MixStreamElementSerializer<>(casted);
		}
	}
}